package org.budo.canal.server;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.budo.canal.instance.spring.CanalInstanceFactoryBean;
import org.budo.canal.message.handler.CanalMessageHandler;
import org.budo.canal.util.MessageUtil;
import org.budo.support.java.concurrent.thread.factory.BudoThreadFactory;
import org.budo.support.javax.sql.util.JdbcUtil;
import org.budo.support.lang.util.CollectionUtil;
import org.budo.support.lang.util.ProcessUtil;
import org.budo.support.spring.util.SpringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;

import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * 会自己去IOC里找ZkClientx; 调用 start 方法启动后，会初始化并扫描所有 CanalInstance 并启动监听
 * 
 * @author limingwei
 */
@Getter
@Setter
@Slf4j
public class BudoCanalServerImpl extends AbstractBudoCanalServer implements DisposableBean {
    private Map<ClientIdentity, ExecutorService> canalGetThreadExecutor = new ConcurrentHashMap<ClientIdentity, ExecutorService>();

    private static final Logger threadLog = LoggerFactory.getLogger("BudoCanalServer_getWithoutAck_thred");

    public BudoCanalServerImpl() {
        log.info("#47 init BudoCanalServerImpl, this=" + this);
    }

    public void start() {
        try {
            this.startDestinations();
        } catch (Throwable e) {
            log.error("#36 canalServer.start error, e=" + e, e);
        }
    }

    /**
     * 启动所有 CanalInstance
     */
    private void startDestinations() {
        String[] instanceNames = this.getApplicationContext().getBeanNamesForType(CanalInstance.class);
        log.info("#58 startDestinations, instanceNames=" + Arrays.toString(instanceNames));

        for (String beanName : instanceNames) {
            String destination = beanName; // 以 beanName 作为 destination
            try {
                this.startDestination(destination);
            } catch (Exception e) { // 一个失败后，启动下一个
                log.error("#49 startGetWithoutAckThread_1 for " + destination + " error, e=" + e, e);
            }
        }
    }

    protected void onServerRunningMonitorProcessActiveEnter(CanalServerWithEmbedded canalServer, String destination) {
        log.info("#76 serverRunningMonitorProcessActiveEnter, start destination=" + destination);

        canalServer.start(destination);

        log.info("#80 destination=" + destination + ", canalServer=" + canalServer);
        this.startDestination(destination);
    }

    /**
     * 启动一个 CanalInstance
     */
    public void startDestination(String destination) {
        CanalServerWithEmbedded canalServer = this.canalServer();
        Map<String, CanalInstance> canalInstances = canalServer.getCanalInstances();
        log.info("#83 startDestination, canalServer=" + canalServer + ", destination=" + destination + ", pid=" + ProcessUtil.getCurrentProcessId() + ", canalInstances=" + canalInstances);

        // 未启动
        boolean contains = canalInstances.containsKey(destination);
        boolean started = canalServer.isStart(destination);

        if (!contains || !started) {
            ServerRunningMonitor serverRunningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
            log.info("#91 serverRunningMonitor.start, destination=" + destination + ", canalServer=" + canalServer + ", canalInstances=" + canalInstances);
            serverRunningMonitor.start(); // 如果数据库连接不上，这里会报错
        }

        log.info("#95 after start, destination=" + destination + ", contains=" + contains + ", started=" + started);

        // 检查,否则不启动监听,for集群环境
        if (!canalServer.isStart(destination)) {
            log.warn("#99 destination not started, destination=" + destination + ", canalServer=" + canalServer + ", canalInstances=" + canalInstances);
            return;
        }

        short clientId = 1001; // @see https://github.com/alibaba/canal/wiki/ClientAPI
        ClientIdentity clientIdentity = new ClientIdentity(destination, clientId);

        canalServer.subscribe(clientIdentity); // subscribe应可重复调用@seecom.alibaba.otter.canal.meta.MemoryMetaManager#subscribe(ClientIdentity)
        log.info("#107 after subscribe, canalServer=" + canalServer + ", clientIdentity=" + clientIdentity);

        this.startGetWithoutAckThread_2(canalServer, clientIdentity);
    }

    /**
     * 拉取 binlog
     */
    private void getWithoutAck(CanalServerWithEmbedded canalServer, ClientIdentity clientIdentity, CanalMessageHandler canalMessageHandler, int batchSize, Long timeout) {
        Message message = null;
        try {
            message = canalServer.getWithoutAck(clientIdentity, batchSize, timeout, TimeUnit.MILLISECONDS);
            if (message == null || message.getId() == -1 || CollectionUtil.isEmpty(message.getEntries())) {
                threadLog.debug("#126 message={}, clientIdentity={}", message, clientIdentity);
                return;
            }

            canalMessageHandler.handleMessage(message, clientIdentity);
            canalServer.ack(clientIdentity, message.getId()); // ack
        } catch (Throwable e) {
            if (JdbcUtil.isTableNotExist(e) // MySQLSyntaxErrorException: Table 'tjk_dw.tj_warning_client_system_msg' doesn't exist
                    || JdbcUtil.isDataTooLong(e) // MysqlDataTruncation: Data truncation: Data too long for column 'open_store' at row 1
                    || JdbcUtil.isIncorrectStringValue(e)) { // MySQLSyntaxErrorException: Table 'tjk_dw.tj_warning_client_system_msg' doesn't exist
                log.error("#137 getWithoutAck error, rollback now, message=" + message + ", e=" + e, e);
            } else {
                log.error("#139 getWithoutAck error, rollback now, message=" + message + "e=" + e, e);
            }

            if (null != message) {
                canalServer.rollback(clientIdentity, message.getId()); // rollback

                log.info("#137 rollback successfully, clientIdentity={}, message={}", clientIdentity, MessageUtil.toSimpleString(message));
            }
        }
    }

    private void startGetWithoutAckThread_2(final CanalServerWithEmbedded canalServer, final ClientIdentity clientIdentity) {
        ExecutorService executorService = this.canalGetThreadExecutor.get(clientIdentity);
        if (null != executorService) {
            log.warn("#145 canalGetThreadExecutor exists, executorService=" + executorService + ", clientIdentity=" + clientIdentity + ", canalServer=" + canalServer);
            return;
        }

        log.info("#149 startGetWithoutAckThread_2, canalServer=" + canalServer + ", clientIdentity=" + clientIdentity);

        // 启线程
        String threadNamePrefix = "CanalGetThread-" + clientIdentity.getDestination();
        ExecutorService executorServiceNew = Executors.newSingleThreadExecutor(new BudoThreadFactory(threadNamePrefix));

        // put,while时会判断
        canalGetThreadExecutor.put(clientIdentity, executorServiceNew);

        CanalInstanceFactoryBean canalInstanceFactoryBean = (CanalInstanceFactoryBean) SpringUtil.getBeanCached(this.getApplicationContext(), "&" + clientIdentity.getDestination());

        final Integer batchSize = canalInstanceFactoryBean.getBatchSize();
        final Long timeout = canalInstanceFactoryBean.getTimeout();
        final CanalMessageHandler canalMessageHandler = canalInstanceFactoryBean.getCanalMessageHandler();

        // submit
        executorServiceNew.submit(new Runnable() {
            public void run() {
                log.info("#159 before while true, canalServer=" + canalServer + ", clientIdentity=" + clientIdentity + ", this=" + this);

                ExecutorService executorService = BudoCanalServerImpl.this.getCanalGetThreadExecutor().get(clientIdentity);
                while (null != executorService) { // 判断是否put,remove后就停
                    BudoCanalServerImpl.this.getWithoutAck(canalServer, clientIdentity, canalMessageHandler, batchSize, timeout);
                }

                log.warn("#166 after while true, canalServer=" + canalServer + ", clientIdentity=" + clientIdentity + ", this=" + this);
            }
        });
    }

    @Override
    public void stopThread() {
        this.shutdown();
    }

    @Override
    public void destroy() throws Exception {
        this.shutdown();
    }

    protected void shutdown() {
        log.info("#174 shutdown ing, this=" + this);

        Iterator<Entry<ClientIdentity, ExecutorService>> iterator = this.getCanalGetThreadExecutor().entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<ClientIdentity, ExecutorService> entry = iterator.next();
            ClientIdentity clientIdentity = entry.getKey();
            ExecutorService executorService = entry.getValue();

            log.info("#177 shutdown ing, executorService=" + executorService + ", clientIdentity=" + clientIdentity);

            iterator.remove(); // 移除导致循环停止
            log.info("#177 removed, executorService=" + executorService + ", clientIdentity=" + clientIdentity);

            // 尝试 shutdown 100 次
            for (int i = 0; i < 100; i++) {
                executorService.shutdown();

                if (executorService.isShutdown()) {
                    log.info("#186 isShutdown break, executorService=" + executorService + ", clientIdentity=" + clientIdentity);
                    break; // 跳出内层循环,处理下一个外循环
                }

                // shutdown 过程中
                log.info("#190 executorService=" + executorService + ", shutdown ing");
            }
        }

        log.info("#199 done shutdown, this=" + this);
    }
}